package to_hbase

import com.alibaba.fastjson.{JSON, JSONObject}
import org.apache.hadoop.hbase.client.Put
import org.apache.hadoop.hbase.io.ImmutableBytesWritable
import org.apache.hadoop.hbase.util.Bytes
import org.apache.hadoop.mapreduce.Job
import org.apache.spark.rdd.RDD
import org.apache.spark.{ SparkConf, SparkContext}
import scala.collection.JavaConverters.asScalaSetConverter
import scala.collection.mutable
import common.Utils

object Map04_Detail {
  def main(args: Array[String]): Unit = {
    val inpath="file:///Users/kelisiduofu/IdeaProjects/zhiwang_file/jiaoyu/Detaildic/part-00000"
    val conf=new SparkConf().setAppName("Map04_Detail").setMaster("local[*]")
    val sc=new SparkContext(conf)
    val lines=sc.textFile(inpath)
//    val table_name="detail"
    val table_name="DetailMapInfos"
    val job: Job = Utils.get_job(sc, table_name)
    val rdd: RDD[(ImmutableBytesWritable, Put)] = lines.map(line => {
      val map = new mutable.HashMap[String, String]()
      val obj: JSONObject = JSON.parseObject(line)
      obj.keySet().asScala.foreach(key => {
        val value = obj.getString(key)
        map += (key -> value)
      })
      val key = map("WebCode") + "_" + map("SourceCode") //注一
      val put = new Put(Bytes.toBytes(key))
      //addFamilys("DetailMapInfos", map("ConceptId")) 注二
      put.addColumn(Bytes.toBytes(map("ConceptId")), Bytes.toBytes(map("DetailId")), Bytes.toBytes(map("DetailName"))) //注一
      (new ImmutableBytesWritable, put)
    })
    rdd.saveAsNewAPIHadoopDataset(job.getConfiguration)
    sc.stop()
  }
}




/*
注一：
下面全为空，所以可以不用判断，直接拿来用
db.getCollection("DetailDic_IBRD_20220420").find({"ConceptId":  {"$exists":false}})
db.getCollection("DetailDic_IBRD_20220420").find({"DetailId":  {"$exists":false}})
db.getCollection("DetailDic_IBRD_20220420").find({"DetailName":  {"$exists":false}})
db.getCollection("DetailDic_IBRD_20220420").find({"SourceCode":  {"$exists":false}})
db.getCollection("DetailDic_IBRD_20220420").find({"WebCode":  {"$exists":false}})
*/

/*
注二
一直报错,因为它设置了表不可用，又设置可用，导致表不可用
改为手动添加列族
db.getCollection('DetailDic_IBRD_20220420').distinct("ConceptId")将查到的结果手动添加到列族
alter '表名', '列族名'
*/
